package com.yf.dxapp.module.app.redis;

import cn.hutool.core.util.TypeUtil;
import cn.hutool.json.JSONUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;

import java.lang.reflect.Type;

@Slf4j
public abstract class AbstractRedisStreamListener<T> implements StreamListener<String, ObjectRecord<String, String>> {

    private final Class<T> messageType;

    @Getter
    private final String streamKey;

    @SneakyThrows
    protected AbstractRedisStreamListener() {
        Type type = TypeUtil.getTypeArgument(getClass(), 0);
        if (type == null) {
            throw new IllegalStateException(String.format("类型(%s) 需要设置消息类型", getClass().getName()));
        }
        this.messageType = (Class<T>) type;
        //统一都用类名
        this.streamKey = messageType.getSimpleName();
    }

    @Setter
    private RedisTemplate redisTemplate;

    /**
     * Redis 消费者分组，默认使用 spring.application.name 名字
     */
    @Value("${spring.application.name}")
    @Getter
    private String group;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        T messageObject = JSONUtil.toBean(message.getValue(), messageType);
        try {
            this.onMessage(messageObject);
            //然后确认消息
            redisTemplate.opsForStream().acknowledge(group, message);
        } catch (Exception e) {
            log.error("处理消息失败:{}", message.getClass().getSimpleName(), e);
        }
    }

    public abstract void onMessage(T message);


}
